package com.lagou.rpc.consumer.server;

import cn.hutool.core.map.MapUtil;
import com.lagou.rpc.api.IUserService;
import com.lagou.rpc.api.NodeChangeListener;
import com.lagou.rpc.api.RpcRegistryCenter;
import com.lagou.rpc.common.RequestMetrics;
import com.lagou.rpc.factory.RpcRegistryCenterFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

/**
 * @param
 * @author oyz
 * @version 1.0
 * @description: TODO 用来获取服务端提供的节点信息
 * @date 2022/2/28 15:59
 */
@Service
public class RpcConsumer implements NodeChangeListener {

    /**
     * 定义一个map用来存储客户端连接地址信息
     *  key: com.lagou.edu.api.UserService/127.0.0.1:8899
     *  value: 127.0.0.1:8899
     */
    private  Map<String,String> RPC_CLIEENT_POOL = new ConcurrentHashMap<>();

    RpcRegistryCenter registryCenter;

    public RpcConsumer(){
        // 用来存储消费者实例的map
        Map<String,Object> zkServiceMap = new ConcurrentHashMap<>();
        zkServiceMap.put(IUserService.class.getName(),IUserService.class);
        if (null== registryCenter){
            registryCenter = new RpcRegistryCenterFactory().getObject();
        }
        zkServiceMap.entrySet().forEach(new Consumer<Map.Entry<String, Object>>() {
            @Override
            public void accept(Map.Entry<String, Object> stringObjectEntry) {
                // /lg-edu-rpc/com.lagou.rpc.api.IUserService/provider
                String serviceName = stringObjectEntry.getKey();
                // 发现/lg-edu-rpc/com.lagou.rpc.api.IUserService/provider 下面的子节点服务地址127.0.0.1:8899组合列表
                List<String> discovery = registryCenter.discovery(serviceName);
                // 遍历服务器需要连接的节点
                for (String connectString : discovery) {
                   RPC_CLIEENT_POOL.put(serviceName+"/"+connectString,connectString);

                }
            }
        });
        registryCenter.addListener(this);
    }



    /**
     * @description: TODO 用于处理系统节点变更的信息
     * @author oyz
     * @param
     * @date 2022/3/1 10:13
     * @version 1.0
     */
    @Override
    public void notify(String service, List<String> serviceList, PathChildrenCacheEvent event) {
        // 节点变更的时间类型
        PathChildrenCacheEvent.Type type = event.getType();
        // 节点变更数据
        ChildData data = event.getData();
        if(null == data){
            return;
        }
        System.out.println("收到节点变更通知:事件类型"+ type + "--服务"+ service + "节点为:"+serviceList);
        // 获取变更节点
        String path = data.getPath();
        // 127.0.0.1:2181
        String substring = path.substring(path.lastIndexOf("/") + 1);
        String key = service + "/" + substring;
        // 判断节点类型(是否添加一个孩子)
        if(PathChildrenCacheEvent.Type.CHILD_ADDED.equals(event)
                ||PathChildrenCacheEvent.Type.CONNECTION_RECONNECTED.equals(event)){
            // 重新放入
            RPC_CLIEENT_POOL.put(key,substring);
            System.out.println("新增加的节点:"+key);
        }
        // 移除节点
        if(PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(event)||
        PathChildrenCacheEvent.Type.CONNECTION_LOST.equals(event)||
                PathChildrenCacheEvent.Type.CONNECTION_SUSPENDED.equals(event)){
            if(!MapUtil.isEmpty(RPC_CLIEENT_POOL)){
                RPC_CLIEENT_POOL.remove(key);
                System.out.println("移除节点:"+key);
            }
            String[] split = substring.split(":");
            RequestMetrics.getInstance().removeNode(split[0],Integer.parseInt(split[1]));
        }
    }


    public Map<String,String> getRpcClieentPool(){
        return RPC_CLIEENT_POOL;
    }
}
